package org.example.dwd;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.streaming.api.scala.DataStream;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.example.util.KafkaUtil;

import java.net.MalformedURLException;
import java.net.URL;

/**
 * @description:
 * @author: 谢超
 * @date: 2024/12/5 9:33
 */
@Slf4j
public class DwdShortLinkLogApp {

    /**
     * 定义topic
     */
    public static final String SOURCE_TOPIC = "ods_link_visit_topic";

    /**
     * 定义消费者组id
     */
    public static final String GROUP_ID = "dwd_short_link_group";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);

//        DataStream<String> ds = env.addSource(kafkaConsumer);

        DataStream<String> ds = env.socketTextStream("127.0.0.1", 8888, '\n', 0);
        ds.print();

        env.execute();
    }

    public static String getReferer(JSONObject jsonObject) {
        JSONObject data = jsonObject.getJSONObject("data");
        if (data.containsKey("referer")) {
            String referer = data.getString("referer");
            if (StringUtils.isNotBlank(referer)) {
                try{
                    URL url = new URL(referer);
                    return url.getHost();
                } catch (MalformedURLException e) {
                    e.printStackTrace();
                }
            }
        }
        return "";
    }
}
